package com.study.chapter11;

import com.study.chapter05.source.ClickSource;
import com.study.entity.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Description:
 * @Author: LiuQun
 * @Date: 2022/8/22 22:34
 */
public class TimeAndWindowTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tabEnv = StreamTableEnvironment.create(env);

        //1.在创建表的DDL中直接定义时间属性：事件时间
        String createDDL = " CREATE TABLE click_table (" +
                " `user` STRING, " +
                " url STRING, " +
                " ts BIGINT, " +
                " event_time AS TO_TIMESTAMP( FROM_UNIXTIME( ts / 1000 ) ), " +   //将ts时间戳转换成timestamp类型
                " WATERMARK FOR event_time AS event_time - INTERVAL '1' SECOND " +  //设置水位线的时间间隔为1s
                " ) WITH ( " +
                " 'connector' = 'filesystem', " +   //指定连接器为文件
                " 'path' = 'input/cart.txt', " +    //指定文件路径
                " 'format' = 'csv' " +              //指定格式
                " ) ";

        tabEnv.executeSql(createDDL);


        //2.在流转换成Table的时候定义时间属性，使用rowtime()：事件时间
        SingleOutputStreamOperator<Event> eventStream = env.addSource(new ClickSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                                    @Override
                                    public long extractTimestamp(Event element, long recordTimestamp) {
                                        return element.timestamp;
                                    }
                                })
                );

        Table table = tabEnv.fromDataStream(eventStream, $("user"), $("url"),
                $("timestamp").as("ts"), $("event_time").rowtime());



        //打印当前的表结构
        table.printSchema();

        //3.处理时间：在创建表的DDL中定义
        String createDDL2 = " CREATE TABLE click_tab (" +
                " `user` STRING, " +
                " url STRING, " +
                " ts BIGINT, " +
                " event_time as PROCTIME() " +   //调用系统内置的函数PROCTIME()获取当前的处理时间属性
                " ) WITH ( " +
                " 'connector' = 'filesystem', " +   //指定连接器为文件
                " 'path' = 'input/cart.txt', " +    //指定文件路径
                " 'format' = 'csv' " +              //指定格式
                " ) ";

        tabEnv.executeSql(createDDL2);


        //4.处理时间：在数据流转换为表的时候定义
        Table table2 = tabEnv.fromDataStream(eventStream, $("user"), $("url"),
                $("timestamp").as("ts"), $("et").proctime());

        table2.printSchema();

        //5.分组聚合
        Table aggTab = tabEnv.sqlQuery("select user,count(1) as num from click_table group by user");
        // tabEnv.toChangelogStream(aggTab).print("aggTab");

        //5.分组窗口聚合(老版本)
        Table windAggTab = tabEnv.sqlQuery(" select " +
                " user,count(1) as num, " +
                " TUMBLE_END(event_time, INTERVAL '10' SECOND ) AS entT " +
                " from click_table " +
                " group by " +
                " user, " +
                " TUMBLE(event_time, INTERVAL '10' SECOND ) "
        );
        tabEnv.toDataStream(windAggTab).print("windAggTab");

        //6.窗口聚合
        //6.1滚动窗口
        Table tumbleWindTab = tabEnv.sqlQuery("select user,count(1) as num," +
                "  window_end as endT " +
                " from TABLE(" +        //使用TABLE()函数
                "   TUMBLE( TABLE click_table, DESCRIPTOR(event_time), INTERVAL '10' SECOND)" +
                " )" +
                " GROUP BY user,window_end,window_start"
        );

        tabEnv.toDataStream(tumbleWindTab).print("tumbleWindTab");

        //6.2 滑动窗口
        Table hotWindTab = tabEnv.sqlQuery("select user,count(1) as num," +
                "  window_end as endT " +
                " from TABLE(" +        //使用TABLE()函数
                "   HOP( TABLE click_table, DESCRIPTOR(event_time), INTERVAL '5' SECOND,INTERVAL '10' SECOND)" +
                " )" +
                " GROUP BY user,window_end,window_start"
        );
        tabEnv.toDataStream(hotWindTab).print("hotWindTab");

        // 6.3 累积窗口
        Table accWindTab = tabEnv.sqlQuery("select user,count(1) as num," +
                "  window_end as endT " +
                " from TABLE(" +        //使用TABLE()函数
                "   CUMULATE( TABLE click_table, DESCRIPTOR(event_time), INTERVAL '5' SECOND,INTERVAL '10' SECOND)" +
                " )" +
                " GROUP BY user,window_end,window_start"
        );
        tabEnv.toDataStream(accWindTab).print("accWindTab");


        //7.开窗聚合
        Table overTab = tabEnv.sqlQuery(" select user,ts,COUNT(url) OVER(" +
                "   PARTITION BY user" +
                "   ORDER BY event_time" +
                "   RANGE BETWEEN INTERVAL '1' HOUR PRECEDING AND CURRENT ROW  " +
                "   ) AS cnt" +
                " FROM click_table" +
                " ");
        tabEnv.toDataStream(overTab).print("overTab");


        env.execute();

    }
}
